/*
 *Copyright 2014 DDPush
 *Author: AndyKwok(in English) GuoZhengzhu(in Chinese)
 *Email: ddpush@126.com
 *

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

*/
package org.ddpush.im.v1.client.appuser;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

/**
 *@ http://blog.csdn.net/brok1n/article/details/44531867
 * TCP客户端基类
 */
public abstract class TCPClientBase implements Runnable {
	/** 超时时间 单位 秒 */
	protected static int connectTimeout = 10;
	/** 套接字缓冲 */
	protected SocketChannel channel;
	/** 最后一次发送消息包得时间 */
	protected long lastSent = 0;
	/** 远程服务器端口 */
	protected int remotePort = 9966;
	/** 应用id */
	protected int appid = 1;
	/** uuid */
	protected byte[] uuid;
	/** 远程服务器地址 */
	protected String remoteAddress = null;
	/** 并发消息队列 */
	protected ConcurrentLinkedQueue<Message> mq = new ConcurrentLinkedQueue<Message>();
	/** 消息队列接收到得消息计数器 */
	protected AtomicLong queueIn = new AtomicLong(0);
	/** 消息队列取出消息计数器 */
	protected AtomicLong queueOut = new AtomicLong(0);
	/** 缓冲区大小 */
	protected int bufferSize = 1024;
	/** 心跳包间隔时间 单位（秒 ） */
	protected int heartbeatInterval = 50;
	/** 用来存放消息数据的byte[] */
	protected byte[] bufferArray;
	/** 处理消息数据的缓冲区 */
	protected ByteBuffer buffer;
	/** 是否需要重置连接 */
	protected boolean needReset = true;
	/** 启动状态标识 */
	protected boolean started = false;
	/** 停止状态标识 */
	protected boolean stoped = false;
	/** 当前TCPClientBase线程 */
	protected Thread receiverT;

	/** 工作线程类 （消息处理类） */
	protected Worker worker;
	/** 工作线程 */
	protected Thread workerT;
	/** 发送的数据包个数 */
	private long sentPackets;
	/** 接收到得数据包个数 */
	private long receivedPackets;

	/** TCP客户端初始化 */
	public TCPClientBase(byte[] uuid, int appid, String serverAddr, int serverPort, int connectTimeout)
			throws Exception {
		if (uuid == null || uuid.length != 16) {
			throw new java.lang.IllegalArgumentException("uuid byte array must be not null and length of 16 bytes");
		}
		if (appid < 1 || appid > 255) {
			throw new java.lang.IllegalArgumentException("appid must be from 1 to 255");
		}
		if (serverAddr == null || serverAddr.trim().length() == 0) {
			throw new java.lang.IllegalArgumentException("server address illegal: " + serverAddr);
		}

		this.uuid = uuid;
		this.appid = appid;
		this.remoteAddress = serverAddr;
		this.remotePort = serverPort;
		TCPClientBase.connectTimeout = connectTimeout;
	}

	/** 入队 将消息添加到消息队列 */
	protected boolean enqueue(Message message) {
		boolean result = mq.add(message);
		if (result == true) {
			queueIn.addAndGet(1);
		}
		return result;
	}

	/** 出队 将消息从消息队列中取出 取出最早放入的消息 */
	protected Message dequeue() {
		Message m = mq.poll();
		if (m != null) {
			queueOut.addAndGet(1);
		}
		return m;
	}

	/** 初始化 数据块以及缓冲 */
	private synchronized void init() {
		bufferArray = new byte[bufferSize];
		buffer = ByteBuffer.wrap(bufferArray);
		buffer.limit(Message.SERVER_MESSAGE_MIN_LENGTH);
	}

	/** socket重置 和服务器的连接中断了就需要重新连接 */
	protected synchronized void reset() throws Exception {
		// 如果连接被关闭了。就重新创建连接
		if (needReset == false) {
			return;
		}
		// 如果socket通道 != null 就把socket通道销毁
		if (channel != null) {
			try {
				channel.socket().close();
			} catch (Exception e) {
			}
			try {
				channel.close();
			} catch (Exception e) {
			}
		}
		/** 检测网络连接状态 如果网络连接状态正常就开始创建连接远程服务器 */
		if (hasNetworkConnection() == true) {
			channel = SocketChannel.open();
			// 当前socket连接被设置为 阻塞模式
			channel.configureBlocking(true);
			channel.socket().connect(new InetSocketAddress(remoteAddress, remotePort), 1000 * connectTimeout);
			channel.socket().setSoTimeout(1000 * 5);
			needReset = false;
		} else {
			try {
				Thread.sleep(1000);
			} catch (Exception e) {
			}
		}
	}

	/** 启动TCPClientBase */
	public synchronized void start() throws Exception {
		if (this.started == true) {
			return;
		}
		// 初始化
		this.init();
		// 接收线程
		receiverT = new Thread(this, "DDPUSH-TCP-CLIENT-RECEIVER");
		// 设置为守护线程 主线程结束它就结束
		receiverT.setDaemon(true);
		synchronized (receiverT) {
			receiverT.start();
			receiverT.wait();
		}
		// 工作线程 发送心跳包 以及处理接收到的消息
		worker = new Worker();
		workerT = new Thread(worker, "DDPUSH-TCP-CLIENT-WORKER");
		workerT.setDaemon(true);
		synchronized (workerT) {
			workerT.start();
			workerT.wait();
		}

		this.started = true;
	}

	/**
	 * 停止TCPClientBase
	 */
	public synchronized void stop() {
		// 改变客户端状态
		stoped = true;
		// 如果socket通道 != null 就关闭他 并将他置 null
		if (channel != null) {
			try {
				channel.socket().close();
			} catch (Exception e) {
			}
			;
			try {
				channel.close();
			} catch (Exception e) {
			}
			;
		}
		channel = null;

		// 如果当前线程 != null 就中断当前线程
		if (receiverT != null) {
			try {
				receiverT.interrupt();
			} catch (Exception e) {
			}
		}

		// 如果worker工作线程 != null 就中断工作线程
		if (workerT != null) {
			try {
				workerT.interrupt();
			} catch (Exception e) {
			}
		}
	}

	/** 当前TCPClientBase线程体 */
	public void run() {
		// 唤醒当前TCPClientBase的所有处于wait状态的线程
		synchronized (receiverT) {
			receiverT.notifyAll();
		}
		// 如果 当前client状态 stoped != true 状态。就开始一直工作
		while (stoped == false) {
			try {
				// 如果网络状态不正常 无网络连接 的时候。就一直跳出循环。不连接服务器 不处理数据。直到网络连接正常
				if (hasNetworkConnection() == false) {
					try {
						trySystemSleep();
						Thread.sleep(1000);
					} catch (Exception e) {
					}
					continue;
				}
				// 检测连接是否被关闭
				reset();

				// 接收服务器发送来得数据。并存放到消息队列中
				// 然后唤醒work线程开始处理消息包
				receiveData();
			} catch (java.net.SocketTimeoutException e) {

			} catch (java.nio.channels.ClosedChannelException e) {
				// 连接被关闭
				this.needReset = true;
			} catch (Exception e) {
				e.printStackTrace();
				// 连接被关闭
				this.needReset = true;
			} catch (Throwable t) {
				t.printStackTrace();
				this.needReset = true;
			} finally {

				// 如果连接被关闭 就休息一会儿
				if (needReset == true) {
					try {
						trySystemSleep();
						Thread.sleep(1000);
					} catch (Exception e) {
					}
				}
				// 如果消息队列是空得或者网络有问题。就休息一会儿
				if (mq.isEmpty() == true || hasNetworkConnection() == false) {
					try {
						trySystemSleep();
						Thread.sleep(1000);
					} catch (Exception e) {
					}
				}
			}
		}
		// 销毁socket
		if (this.channel != null) {
			try {
				channel.socket().close();
			} catch (Exception e) {
			}
			try {
				channel.close();
			} catch (Exception e) {
			}
			channel = null;
		}
	}

	/**
	 * 发送心跳包
	 * 
	 * @throws Exception
	 */
	private void heartbeat() throws Exception {
		// 当前时间和上次发送数据时间间隔不超过heartbeatInterval秒。就不发送心跳。
		if (System.currentTimeMillis() - lastSent < heartbeatInterval * 1000) {
			return;
		}
		// 创建消息包头，应为这是个心跳包。所以没有消息内容。只有消息包头。所以创建的消息byte[]长度是
		// Message.CLIENT_MESSAGE_MIN_LINGHT
		// 即Client数据包最小长度
		byte[] buffer = new byte[Message.CLIENT_MESSAGE_MIN_LENGTH];
		// 用buffer作存储空间创建一个缓冲区
		// 第一位 消息版本
		// 第二位 appid
		// 第三位 消息类型 MC_0x00 心跳包
		// 第四位 uuid
		// 第五位 0

		ByteBuffer.wrap(buffer).put((byte) Message.version).put((byte) appid).put((byte) Message.CMD_0x00).put(uuid)
				.putChar((char) 0);
		// 发送心跳包
		send(buffer);
	}

	/**
	 * 接收服务器发送的数据
	 * 
	 * @throws Exception
	 */
	private void receiveData() throws Exception {
		// 判断消息包是否可操作 是否达到了上限 limit
		// 整体来说。这个while循环将服务器发送来得消息包分为包头和消息内容两次来读取到buffer中
		while (hasPacket() == false) {
			// 如果buffer没有达到上限 也就是 SERVER消息的最小长度
			// 就从socket通道里读取数据读取放到buffer中，返回值是本次读取了多少数据放到buffer中
			int read = channel.read(buffer);
			// 如果读取的数据长度 < 0 就表明这个连接被关闭了
			if (read < 0) {
				throw new Exception("end of stream");
			}
			// 如果读取结束
			if (hasPacket() == true) {
				break;
			}

			// 如果消息队列为空 或者网络状态 出问题了。就休眠等待
			if (mq.isEmpty() == true || hasNetworkConnection() == false) {
				try {
					trySystemSleep();
					Thread.sleep(1000);
				} catch (Exception e) {
				}
			}
		}
		// 根据bufuer的当前操作位置，创建一个刚好能够容纳buffer中数据的byte[]
		byte[] data = new byte[buffer.position()];
		// byte[] 拷贝 将接受到得数据拷贝到一个新的byte[]中以便
		// 当前的buffer能够继续的接受服务器发送来得消息数据
		System.arraycopy(bufferArray, 0, data, 0, buffer.position());
		// 根据服务器发送来得消息数据。创建一个Message消息数据包
		Message m = new Message(channel.socket().getRemoteSocketAddress(), data);
		// 清空缓冲区
		buffer.clear();
		// 设置缓冲区的操作限制
		buffer.limit(Message.SERVER_MESSAGE_MIN_LENGTH);

		// 如果这个数据包格式错误就丢弃不处理他
		if (m.checkFormat() == false) {
			return;
		}
		// 修改接收到的数据包个数
		this.receivedPackets++;

		// 告诉服务器。我收到的数据包
		this.ackServer(m);

		// 如果是心跳包。就丢弃这个消息
		if (m.getCmd() == Message.CMD_0x00) {
			return;
		}
		// 将接收到得消息放到消息队列
		this.enqueue(m);
		// 唤醒工作线程、开始处理接收到得消息包
		worker.wakeup();
	}

	/**
	 * 判断是否有可以接收的数据*
	 * 
	 * @return
	 */
	private boolean hasPacket() {
		// 如果buffer的上限limit == Message.SERVER_MESSAGE_MIN_LENGHT SERVER消息包最小长度
		if (buffer.limit() == Message.SERVER_MESSAGE_MIN_LENGTH) {
			// 如果buffer未达到上限
			if (buffer.hasRemaining() == true) {
				return false;
			} else {
				// 如果已经读取完成了SERVER数据包的包头。就开始查看数据包中是否有 消息内容
				// 获取数据的内容长度
				int dataLen = (int) ByteBuffer.wrap(bufferArray, Message.SERVER_MESSAGE_MIN_LENGTH - 2, 2).getChar();
				// 是否有消息内容
				if (dataLen == 0) {
					return true;
				} else {
					// 设置buffer的可操作区域为 消息包头长度 + 消息数据长度
					buffer.limit(Message.SERVER_MESSAGE_MIN_LENGTH + dataLen);
					return false;
				}
			}
		} else {
			// buffer是否可操作 是否达到上限limit
			if (buffer.hasRemaining() == true) {
				return false;
			} else {
				return true;
			}
		}
	}

	/**
	 * 向服务器发送 接收到服务器发送的数据包的 应答包 告诉服务器我收到了一个 什么样的数据包
	 * 
	 * @param m
	 * @throws Exception
	 */
	private void ackServer(Message m) throws Exception {
		if (m.getCmd() == Message.CMD_0x10) {
			byte[] buffer = new byte[Message.CLIENT_MESSAGE_MIN_LENGTH];
			ByteBuffer.wrap(buffer).put((byte) Message.version).put((byte) appid).put((byte) Message.CMD_0x10).put(uuid)
					.putChar((char) 0);
			send(buffer);
		}
		if (m.getCmd() == Message.CMD_0x11) {
			byte[] buffer = new byte[Message.CLIENT_MESSAGE_MIN_LENGTH + 8];
			byte[] data = m.getData();
			ByteBuffer.wrap(buffer).put((byte) Message.version).put((byte) appid).put((byte) Message.CMD_0x11).put(uuid)
					.putChar((char) 8).put(data, Message.SERVER_MESSAGE_MIN_LENGTH, 8);
			send(buffer);
		}
		if (m.getCmd() == Message.CMD_0x20) {
			byte[] buffer = new byte[Message.CLIENT_MESSAGE_MIN_LENGTH];
			ByteBuffer.wrap(buffer).put((byte) Message.version).put((byte) appid).put((byte) Message.CMD_0x20).put(uuid)
					.putChar((char) 0);
			send(buffer);
		}
	}

	/**
	 * 向服务器发送消息
	 * 
	 * @param data
	 * @throws Exception
	 */
	private void send(byte[] data) throws Exception {
		// 数据为空或者socket通道没有准备好就不发送消息
		if (data == null) {
			return;
		}
		if (channel == null || channel.isOpen() == false) {
			return;
		}
		// 缓冲区包裹数据
		ByteBuffer bb = ByteBuffer.wrap(data);
		// 判断当前buffer操作位置是否在可操作限制之内。如果当前位置可以操作。就向cannel写一个字节序列
		while (bb.hasRemaining()) {
			channel.write(bb);
		}
		// 将缓冲区内的数据取出并发送、
		channel.socket().getOutputStream().flush();
		// 记录最后一次(本次发送消息的时间。)
		lastSent = System.currentTimeMillis();
		// 更新发送的数据包个数
		this.sentPackets++;
	}

	/** 获取消息发送次数 */
	public long getSentPackets() {
		return this.sentPackets;
	}

	/** 得到消息接收次数 */
	public long getReceivedPackets() {
		return this.receivedPackets;
	}

	/** 得到最后一次发送消息的时间 */
	public long getLastHeartbeatTime() {
		return lastSent;
	}

	/**
	 * send heart beat every given seconds </br>
	 * 设置心跳时间间隔
	 */
	public void setHeartbeatInterval(int second) {
		if (second <= 0) {
			return;
		}
		this.heartbeatInterval = second;
	}

	/**
	 * 获取心跳时间间隔 单位（秒）
	 * 
	 * @return
	 */
	public int getHeartbeatInterval() {
		return this.heartbeatInterval;
	}

	/**
	 * 检测网络状态
	 * 
	 * @return
	 */
	public abstract boolean hasNetworkConnection();

	/**
	 * 休眠
	 */
	public abstract void trySystemSleep();

	/**
	 * 收到消息的 消息处理 回调
	 * 
	 * @param message
	 */
	public abstract void onPushMessage(Message message);

	/**
	 * 发送心跳包以及处理收到的消息包 的线程
	 * 
	 * @author lilei
	 *
	 */
	class Worker implements Runnable {
		public void run() {
			// 唤醒工作线程
			synchronized (workerT) {
				workerT.notifyAll();
			}
			// 如果当前Client是正在运行的状态。否则，就结束客户端工作线程
			while (stoped == false) {
				try {
					// 发送心跳包。
					heartbeat();
					// 处理Message消息
					handleEvent();
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// 休眠一秒
					waitMsg();
				}
			}
		}

		/**
		 * 休眠1s
		 */
		private void waitMsg() {
			synchronized (this) {
				try {
					this.wait(1000);
				} catch (java.lang.InterruptedException e) {

				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

		/**
		 * //唤醒Worker里的所有wait状态的线程退出wait状态。
		 */
		private void wakeup() {
			synchronized (this) {
				this.notifyAll();
			}
		}

		/**
		 * //处理Message消息
		 * 
		 * @throws Exception
		 */
		private void handleEvent() throws Exception {
			Message m = null;
			// 这里的死循环是为了要在这里处理完消息队列中得所有消息
			while (true) {
				// 取出一个消息
				m = dequeue();
				// 如果这个取出的消息为null就表明这个消息队列里没有消息了
				if (m == null) {
					return;
				}
				// 检测这个消息的格式是否正确，如果不正确就丢弃当前消息继续处理下一个消息
				if (m.checkFormat() == false) {
					continue;
				}

				// real work here 收到一个消息。在这里回调消息处理的函数
				// 将当前消息传递给消息处理函数
				onPushMessage(m);
			}
			// finish work here, such as release wake lock
		}

	}

}
